package net.zhangyun.reactor;

import net.zhangyun.utils.MyConsole;
import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;

public class BackPressureMain {

    public static void main(String[] args) {
        BackPressureFluxSample();
    }

    private static void BackPressureFluxSample() {
        Flux.range(100,10)
                .doOnRequest(MyConsole::printMessage)
                .subscribe(new BaseSubscriber<Integer>() {
                    @Override
                    protected void hookOnSubscribe(Subscription subscription) {
                        request(5);
                        //super.hookOnSubscribe(subscription);
                        //By default Long.MAX_VALUE is supplied which means sends all data
                    }

                    @Override
                    protected void hookOnNext(Integer value) {
                        MyConsole.printSingle(value);

                        {
                            //cancel(); //If we cancel, it stops immediately
                            //MyConsole.printMessage("After canceling Flux...");
                            //request(5); //Of no use anymore
                        }

                        {
                            //request(1); //If we continue to request, the stream continues
                        }
                    }

                    @Override
                    protected void hookOnComplete() {
                        MyConsole.printMessage("Done");
                    }

                    @Override
                    protected void hookOnError(Throwable throwable) {
                        MyConsole.printMessage(throwable);
                    }

                    @Override
                    protected void hookOnCancel() {
                        MyConsole.printMessage("Canceled");
                    }
                });
    }
}
